package com.xiaomaoguai;

import com.xiaomaoguai.entity.Student;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author WeiHui-Z
 * @version v1.0.0
 * @date 2019/8/30 19:37
 * @since JDK 1.8
 */
public class SinkToMysql extends RichSinkFunction<Student> {

	PreparedStatement ps;
	private Connection connection;

	/**
	 * open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接
	 *
	 * @param parameters
	 * @throws Exception
	 */
	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		connection = getConnection();
		String sql = "insert into Student(id, name, password, age) values(?, ?, ?, ?);";
		ps = this.connection.prepareStatement(sql);
	}

	@Override
	public void close() throws Exception {
		super.close();
		//关闭连接和释放资源
		if (connection != null) {
			connection.close();
		}
		if (ps != null) {
			ps.close();
		}
	}

	/**
	 * 每条数据的插入都要调用一次 invoke() 方法
	 *
	 * @param value
	 * @param context
	 * @throws Exception
	 */
	@Override
	public void invoke(Student value, Context context) throws Exception {
		//组装数据，执行插入操作
		ps.setInt(1, value.getId());
		ps.setString(2, value.getName());
		ps.setString(3, value.getPassword());
		ps.setInt(4, value.getAge());
		ps.executeUpdate();
	}

	private static Connection getConnection() {
		Connection con = null;
		try {
			Class.forName("com.mysql.jdbc.Driver");
			con = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "root123456");
		} catch (Exception e) {
			System.out.println("-----------mysql get connection has exception , msg = "+ e.getMessage());
		}
		return con;
	}

}
